#ifndef __M_MSG_H__
#define __M_MSG_H__
#include "../mqCommon/logger.hpp"
#include "../mqCommon/helper.hpp"
#include "../mqCommon/mq_msg.pb.h"
#include <iostream>
#include <memory>
#include <mutex>
#include <unordered_map>
#include <list>
/*
    设计说明：
    1.消息是以数据为单元管理起来的，数据持久化也是以队列为单位————即一个队列的消息存在一个文件中！
    2.除了MessageMapper这个数据持久化管理类外，还设计了两个数据管理类，
      第一个是QueueMessageManager，用于操作一个具体的队列的消息数据，
      第二个是MessageManager，用于管理所有的队列以及它们其中的消息，
      通过队列名可以在MessageManager找到该队列的QueueMessageManager句柄，从而操作该队列管理的消息。
*/
namespace yqx{
    #define DATAFILE_SUBFIX  ".mqd"
    #define TMPFILE_SUBFIX  ".mqd.tmp"
    using MessagePtr = std::shared_ptr<yqx::Message>;
    /*  
        消息持久化模块说明：
        1.消息是以队列为单位管理起来的，即一个队列一个文件，来存储这个队列里的消息数据——避免一个大文件存所有队列的消息，频繁加锁，效率低下
        2.消息的存储没有使用数据库，因为消息不定长且有些消息内容太长。这里使用了普通文件来存储消息，格式是：4字节长度|数据|4字节长度|数据...
        3.删除文件中的消息时并不会把消息真的删除，而只是把valid位置0,所以文件中的垃圾消息会越来越多，当文件中消息超过2000条且有效消息不超过50%，
          启动垃圾回收机制：加载文件中所有有效信息到一个临时文件中，全部成功加载后删除源文件，把临时文件名改称源文件名，返回所有有效消息（每条消息
          都记录当前存储的新位置，用于更新内训中消息的数据内容）
    */
    //消息数据持久化管理类
    class MessageMapper {
        public:
            MessageMapper(std::string& basedir, const std::string& qname):
                _qname(qname){
                if(basedir.back() != '/') basedir.push_back('/');
                _datafile = basedir + qname + DATAFILE_SUBFIX;
                _tmpfile = basedir + qname + TMPFILE_SUBFIX;
                if(FileHelper(basedir).exist() == false)
                    assert(FileHelper::createDirectory(basedir));
                createMsgFile();
            }
            //创建储存消息的文件
            bool createMsgFile(){
                if(FileHelper(_datafile).exist() == true)
                    return true;
                bool res = FileHelper::createFile(_datafile);
                if(res == false){
                    LOG("创建队列数据文件“%s”失败", _datafile.c_str());
                    return false;
                }
                return true;
            }
            //删除储存消息的文件
            void removeMsgFile(){
                FileHelper::removeFile(_datafile);
                FileHelper::removeFile(_tmpfile);
            }
            //向文件新增一条消息
            bool insert(MessagePtr& mp){
                return insertToFile(_datafile, mp);
            }
            //删除文件中一条消息
            bool remove(MessagePtr mp){
                //将消息的有效位置为0
                mp->mutable_payload()->set_valid("0");
                //将消息反序列化
                std::string body = mp->payload().SerializeAsString();
                if(body.size() != mp->length()){
                    LOG("反序列化出错！不能修改文件消息！");
                    return false;
                }
                //将序列化后的消息写入到文件中制定位置（覆盖原有数据）
                FileHelper helper(_datafile);
                int res = helper.write(body.c_str(), mp->offset(), body.size());
                if(res == false){
                    LOG("删除消息时写入失败！");
                    return false;
                }
                return true;
            }
            //垃圾回收
            std::list<MessagePtr> gc(){
                std::list<MessagePtr> res;
                //加载文件中所有有效数据
                bool ret;
                ret = load(res);
                if(ret == false){
                    LOG("加载有效数据失败！");
                    return std::list<MessagePtr>();
                }
                //将有效数据存到临时文件中
                FileHelper::createFile(_tmpfile);
                for(auto& mp : res){
                    ret = insertToFile(_tmpfile, mp);
                    if(ret == false){
                        LOG("向临时文件写入数据失败！");
                        return std::list<MessagePtr>();
                    }
                }
                //删除源文件，修改临时文件名
                ret = FileHelper::removeFile(_datafile);
                if(ret == false){
                    LOG("删除文件失败！");
                    return std::list<MessagePtr>();
                }
                ret = FileHelper(_tmpfile).rename(_datafile);
                if(ret == false){
                    LOG("修改临时文件名称失败！");
                    return std::list<MessagePtr>();
                }
                //返回新的有效数据
                return res;
            }
        private:
            //把有效消息加载到链表里
            bool load(std::list<MessagePtr>& res){
                FileHelper helper(_datafile);
                size_t offset = 0, msg_size;
                size_t fsize = helper.size();
                bool ret;
                while(offset < fsize){
                    //读消息长度
                    ret = helper.read((char*)&msg_size, offset, sizeof(size_t));
                    if(ret == false){
                        LOG("读取消息长度失败！");
                        return false;
                    }
                    offset += sizeof(size_t);
                    //读消息内容
                    std::string msg_body(msg_size, '\0');
                    ret = helper.read(&msg_body[0], offset, msg_size);
                    if(ret == false){
                        LOG("读取消息内容失败！");
                        return false;
                    }
                    offset += msg_size;
                    MessagePtr mp = std::make_shared<Message>();
                    mp->mutable_payload()->ParseFromString(msg_body);
                    if(mp->payload().valid() == "1")
                        res.push_back(mp);
                }
                return true;
            }
            bool insertToFile(const std::string& filename, MessagePtr& mp){
                //把消息反序列化
                std::string body = mp->payload().SerializeAsString();
                //获取文件长度
                FileHelper helper(filename);
                size_t fsize = helper.size();
                size_t msg_size = body.size();
                //将数据写入文件
                bool res = helper.write((char*)&msg_size, fsize, sizeof(size_t));
                if(res == false){
                    LOG("向队列数据文件写入消息长度失败！");
                    return false;
                }
                res = helper.write(body.c_str(), fsize + sizeof(size_t), body.size());
                if(res == false){
                    LOG("向队列数据文件写入消息失败！");
                    return false;
                }
                //更新内存中msg的实际存储位置
                mp->set_offset(fsize + sizeof(size_t));
                mp->set_length(body.size());
                return true;
            }
        private:
            std::string _qname;
            std::string _datafile;
            std::string _tmpfile;

    };
    //每一个队列的消息管理模块
    class QueueMessageManager{
        public:
            using ptr = std::shared_ptr<QueueMessageManager>;
            QueueMessageManager(std::string& basedir, const std::string& qname):
            _qname(qname), _mapper(basedir, qname), _valid_count(0), _total_count(0){}
            //恢复持久化数据
            void recovery(){
                std::unique_lock<std::mutex> lock(_mtx);
                _pending_msgs = _mapper.gc();
                for(auto &msg : _pending_msgs){
                    _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
                }
                _valid_count = _total_count = _pending_msgs.size();
            }
            //向队列添加一个消息
            bool insert(const BasicProperties *bp, const std::string& body, bool queue_is_durable){
                //构造消息对象
                MessagePtr msg = std::make_shared<Message>();
                msg->mutable_payload()->set_body(body);
                if(bp){
                    DeliveryMode mode = queue_is_durable ? bp->delivery_mode() : DeliveryMode::UNDURABLE;
                    msg->mutable_payload()->mutable_properties()->set_id(bp->id());
                    msg->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
                    msg->mutable_payload()->mutable_properties()->set_routing_key(bp->routing_key());
                }else{
                    DeliveryMode mode = queue_is_durable ? DeliveryMode::DURABLE : DeliveryMode::UNDURABLE;
                    msg->mutable_payload()->mutable_properties()->set_id(UUIDHelper::uuid());
                    msg->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
                    msg->mutable_payload()->mutable_properties()->set_routing_key("");
                }
                //消息持久化
                std::unique_lock<std::mutex> lock(_mtx);
                if(msg->payload().properties().delivery_mode() == DeliveryMode::DURABLE){
                    msg->mutable_payload()->set_valid("1");//在持久化存储中表示数据有效
                    bool ret = _mapper.insert(msg);
                    if(ret == false){
                        LOG("消息持久化失败！");
                        return false;
                    }
                    _valid_count++;
                    _total_count++;
                    _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
                }
                //把消息放入待推送链表
                _pending_msgs.push_back(msg);
                return true;
            }
            void remove(const std::string& msg_id){
                std::unique_lock<std::mutex> lock(_mtx);
                auto it = _unAcked_msgs.find(msg_id);
                if(it == _unAcked_msgs.end()){
                    LOG("没有找到要删除的信息");
                    return;
                }
                if(it->second->payload().properties().delivery_mode() == DeliveryMode::DURABLE){
                    _mapper.remove(it->second);
                    _durable_msgs.erase(msg_id);
                    _valid_count--;
                    gc();
                        
                }
                _unAcked_msgs.erase(msg_id);
            }
            //获取队首消息
            MessagePtr front(){
                std::unique_lock<std::mutex> lock(_mtx);
                if(_pending_msgs.size() == 0){
                    return nullptr;
                }
                MessagePtr msg = _pending_msgs.front();
                _pending_msgs.pop_front();
                _unAcked_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
                return msg;
            }
            //清除内存和文件中资源
            void clear(){
                std::unique_lock<std::mutex> lock(_mtx);
                _pending_msgs.clear();
                _unAcked_msgs.clear();
                _durable_msgs.clear();
                _total_count = _valid_count = 0;
                _mapper.removeMsgFile();
            }
        private:
            bool gcCheck(){
                if(_total_count > 2000 && _valid_count * 10 / _total_count < 5)
                    return true;
                return false;
            }
            void gc(){
                if(gcCheck() == false) return;
                std::list<MessagePtr> msgs = _mapper.gc();
                for(auto& msg : msgs){
                    auto it = _durable_msgs.find(msg->payload().properties().id());
                    if(it == _durable_msgs.end()){
                        LOG("垃圾回收后，有一条持久化消息，在内存中没有进行管理！");
                        _pending_msgs.push_back(msg);
                        _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
                        continue;
                    }
                    it->second->set_offset(msg->offset());
                    it->second->set_length(msg->length());
                }
                _valid_count = _total_count = msgs.size();
            }
        private:
            std::mutex _mtx;
            std::string _qname;
            MessageMapper _mapper;
            size_t _valid_count;
            size_t _total_count;
            std::list<MessagePtr> _pending_msgs;//待推送消息链表
            std::unordered_map<std::string, MessagePtr> _durable_msgs;//持久化消息hash
            std::unordered_map<std::string, MessagePtr> _unAcked_msgs;//待确认消息hash

    };
    //总的消息管理模块
    class MessageManager{
        public:
            using ptr = std::shared_ptr<MessageManager>;
            MessageManager(const std::string& basedir):_basedir(basedir){}
            //初始化一个队列的管理句柄，并将其添加到_queue_msgs里
            void initQueueMessage(const std::string& qname){
                QueueMessageManager::ptr qmp;
                {
                    std::unique_lock<std::mutex> lock(_mtx);
                    auto it = _queue_msgs.find(qname);
                    if(it != _queue_msgs.end())
                        return;
                    qmp = std::make_shared<QueueMessageManager>(_basedir, qname);
                    _queue_msgs.insert(std::make_pair(qname, qmp));
                }
                qmp->recovery();
            }
            //从_queue_msgs删除一个队列的管理句柄
            void destroyQueueMessage(const std::string& qname){
                QueueMessageManager::ptr qmp;
                {
                    std::unique_lock<std::mutex> lock(_mtx);
                    auto it = _queue_msgs.find(qname);
                    if(it == _queue_msgs.end())
                        return;
                    qmp = it->second;
                    _queue_msgs.erase(it);
                }
                qmp->clear();
            }
            //新增一条消息
            bool insert(const std::string& qname, BasicProperties* bp, const std::string& body, bool queue_is_durable){
                QueueMessageManager::ptr qmp;
                {
                    std::unique_lock<std::mutex> lock(_mtx);
                    auto it = _queue_msgs.find(qname);
                    if(it == _queue_msgs.end()){
                        LOG("添加消息失败！未找到该队列！");
                        return false;
                    }
                    qmp = it->second;
                }
                return qmp->insert(bp, body, queue_is_durable);
            }
            //取出队首信息
            MessagePtr front(const std::string& qname){
                QueueMessageManager::ptr qmp;
                {
                    std::unique_lock<std::mutex> lock(_mtx);
                    auto it = _queue_msgs.find(qname);
                    if(it == _queue_msgs.end()){
                        LOG("获取队列<%s>队首消息失败：没有找到消息管理句柄！", qname.c_str());
                        return nullptr;
                    }
                    qmp = it->second;
                }
                return qmp->front();
            }
            //确认一条消息
            void ack(const std::string& qname, const std::string& msg_id){
                QueueMessageManager::ptr qmp;
                {
                    std::unique_lock<std::mutex> lock(_mtx);
                    auto it = _queue_msgs.find(qname);
                    if(it == _queue_msgs.end()){
                        LOG("确认队列<%s>的消息<%s>失败！", qname.c_str(), msg_id.c_str());
                        return;
                    }
                    qmp = it->second;
                }
                qmp->remove(msg_id);//确认后就删除
            }
            void clear(){
                std::unique_lock<std::mutex> lock(_mtx);
                for(auto& qmsg : _queue_msgs){
                    qmsg.second->clear();
                }
            }
        private:
            std::mutex _mtx;
            std::string _basedir;
            std::unordered_map<std::string, QueueMessageManager::ptr> _queue_msgs;//所有队列的管理句柄
    };
}
#endif